package com.zlx.rulemk.marketing.service;

import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import com.zlx.rulemk.marketing.beans.EventBean;
import com.zlx.rulemk.marketing.beans.EventCombinationCondition;
import com.zlx.rulemk.marketing.beans.EventCondition;
import com.zlx.rulemk.marketing.constant.ConfigNames;
import com.zlx.rulemk.marketing.dao.ClickHouseQuerier;
import com.zlx.rulemk.marketing.dao.HbaseQuerier;
import com.zlx.rulemk.marketing.dao.StateQuerier;
import com.zlx.rulemk.marketing.dao.TriggerQuerier;
import com.zlx.rulemk.marketing.utils.ConnectionUtils;
import com.zlx.rulemk.marketing.utils.EventUtil;
import com.zlx.rulemk.marketing.utils.FlinkPointTimeUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.java.tuple.Tuple2;

import java.io.IOException;
import java.sql.Connection;
import java.util.Map;

/**
 * @Author: zhanglingxing
 * @CreateTime: 2022-07-08
 * @Description: 提供的服务方法
 * @Version: 1.0
 * TODO 这个类目前的作用太重，逻辑代码太多了 后期可以重构
 */
@Slf4j
public class RulelMatchServiceImpl {

    /**
     * 触发条件查询的dao
     */
    TriggerQuerier triggerQuerier;

    /**
     * ck查询的dao
     */
     ClickHouseQuerier clickHouseQuerier;

    /**
     * hbase画像条件的dao
     */
    HbaseQuerier hbaseQuerier;

    /**
     * flink state 查询的dao
     */
    StateQuerier stateQuerier;

    /**
     * 构造函数
     * @param listState
     * @throws Exception
     */
    public RulelMatchServiceImpl(ListState<EventBean> listState) throws Exception {

        Config config = ConfigFactory.load();

        Connection ckConn = ConnectionUtils.getClickhouseConnection();
        clickHouseQuerier = new ClickHouseQuerier(ckConn);

        org.apache.hadoop.hbase.client.Connection hbaseConn = ConnectionUtils.getHbaseConnection();
        String profileTableName = config.getString(ConfigNames.HBASE_PROFILE_TABLE);
        String profileFamilyName = config.getString(ConfigNames.HBASE_PROFILE_FAMILY);
        hbaseQuerier = new HbaseQuerier(hbaseConn, profileTableName, profileFamilyName);

        stateQuerier = new StateQuerier(listState);

        triggerQuerier = new TriggerQuerier();
    }

    /**
     * 画像条件是否匹配
     * @param profileCondition 画像条件
     * @param userId 用户标识
     * @return 是否匹配
     * @throws IOException 异常
     */
    public boolean matchProfileCondition(Map<String,String> profileCondition, String userId) throws IOException {
        log.debug("开始匹配画像条件。。。。");
        boolean b = hbaseQuerier.queryProfileConditionIsMatch(profileCondition,userId);
        log.debug(">>>>画像条件匹配结束，用户ID:{},匹配结果：{}",userId,b);
        return b;
    }

    /**
     * 触发事件是否匹配
     * @param bean 日志对象
     * @param eventCondition 规则条件
     * @return 是否匹配
     */
    public boolean triggerEventMatchCondition(EventBean bean, EventCondition eventCondition){
        boolean b = triggerQuerier.triggerEventMatchCondition(bean, eventCondition);
        log.debug(">>>>触发事件匹配结束，事件ID:{},匹配结果：{}",bean.getEventId(),b);
        return b;
    }


    /**
     * 计算（一个）行为组合条件是否匹配
     * @param event 日志对象
     * @param combinationCondition （一个）行为组合条件
     * @return 是否匹配
     * @throws Exception
     */
    public boolean matchEventCombinationCondition(EventBean event,EventCombinationCondition combinationCondition)
            throws Exception {

        log.debug("开始匹配行为组合条件：规则字符串：{},规则要求的范围{}-{}",combinationCondition.getMatchPattern(),
                combinationCondition.getTimeRangeStart(),combinationCondition.getTimeRangeEnd());

        // 获取当前事件时间对应的查询分界点
        long segmentPoint = FlinkPointTimeUtils.getFlinkPointTime(event.getTimeStamp());

        // 判断条件的时间区间是否跨分界点
        long conditionStart = combinationCondition.getTimeRangeStart();
        long conditionEnd = combinationCondition.getTimeRangeEnd();

        String keyByValue = event.getKeyByValue();

        // 查state状态 查询的数据全部在state中
        if(conditionStart >= segmentPoint){
            log.debug("【近期查询】规则要求的开始时间：{} > flink切分点：{},查询在flinkState中进行",conditionStart,segmentPoint);
            int count = stateQuerier.queryEventCombinationConditionCount(combinationCondition,conditionStart, conditionEnd);
            boolean f = (count >= combinationCondition.getMinLimit() && count <= combinationCondition.getMaxLimit()) ;
            log.debug(">>>>>近期查询：在FlinkState中查询，返回conut；{},结果：{}",count,f);
            return f;
        }

        // 查clickhouse
        else if(conditionEnd<segmentPoint){
            log.debug("【远期查询】规则要求的结束时间：{} < flink切分点：{},查询在CK中进行",conditionEnd,segmentPoint);
            Tuple2<String, Integer> resTuple = clickHouseQuerier.queryEventCombinationConditionCount(keyByValue, combinationCondition);
            boolean f = (resTuple.f1 >= combinationCondition.getMinLimit() && resTuple.f1 <= combinationCondition.getMaxLimit());
            log.debug(">>>>>远期查询：在CK中查询，返回conut；{},结果：{}",resTuple.f1,f);
            return f;
        }

        // 跨区段查询
        else {
            log.debug("【跨界查询】规则要求开始时间:{} < flink切分时间：{} < 规则要求减速时间:{}",conditionStart,segmentPoint,conditionEnd);
            // 先查一次state，看是否能提前结束
            int count = stateQuerier.queryEventCombinationConditionCount(combinationCondition, segmentPoint, conditionEnd);
            log.debug("跨界查询：事件组合条件:{},区间跨界在state尝试,条件起始点:{},条件结束点:{},分界点:{},结果:{}",combinationCondition.getCacheId(),conditionStart,conditionEnd,segmentPoint,count);
            if(count >= combinationCondition.getMinLimit() && count <= combinationCondition.getMaxLimit()) return true;


            /**
             * 再查一次clickhouse，看是否能提前结束
             * 以便在后续的分段组合查询时，一定是state或ck子段都无法独立满足条件的情况
             */
            Tuple2<String, Integer> resTupleCk = clickHouseQuerier.queryEventCombinationConditionCount(keyByValue, combinationCondition);
            log.debug("跨界查询：事件组合条件:{},区间跨界,在ck中尝试,条件起始点:{},条件结束点:{},分界点:{},结果:{}",combinationCondition.getCacheId(),conditionStart,conditionEnd,segmentPoint,resTupleCk.f1);
            if(resTupleCk.f1 >= combinationCondition.getMinLimit() && resTupleCk.f1 <= combinationCondition.getMaxLimit()) return true;

            /**
             * TODO BUG 此处的调用，违反了buffer工作层级的原则，导致此处会跳过缓存处理
             * 补救思路1：在getEventCombinationConditionStr方法中添加缓存处理逻辑，那将是灾难性的
             * 补救思路2：改造queryEventCombinationConditionCount方法，不仅返回count值，同时携带str返回（此str不能是缓存处理中阶段后的部分结果，而应该是整个str结果）
             */
            // 分段组合查询,先从ck中查询序列字符串,再从state中查询序列字符串,拼接后作为整体匹配正则表达式
            //Tuple2<String, Integer> resTuple = clickHouseQuerier.queryEventCombinationConditionCount(event.getDeviceId(), combinationCondition, conditionStart, segmentPoint, true);
            String str2 = stateQuerier.getEventCombinationConditionStr(combinationCondition, segmentPoint, conditionEnd);

            count = EventUtil.sequenceStrMatchRegexCount(resTupleCk.f0 + str2, combinationCondition.getMatchPattern());
            log.debug("跨界查询：事件组合条件:{},区间跨界在分段组合查询,条件起始点:{},条件结束点:{},分界点:{},str1:{},str2{},总结果:{}",combinationCondition.getCacheId(),conditionStart,conditionEnd,segmentPoint,resTupleCk.f0,str2,count);

            // 判断是否匹配成功
            return count >= combinationCondition.getMinLimit() && count <= combinationCondition.getMaxLimit();
        }

    }

}
